package beautiful.butterfly.drds.data_exchange.mysql;


import common.plugin.Consume;
import common.util.Configuration;
import rdbms.writer.Key;
import rdbms.writer.Writer;

import java.util.List;


//TODO writeProxy
public class MysqlWriter extends common.spi.Writer
{


    public static class Job extends common.spi.Writer.Job
    {
        private Configuration originalConfig = null;
        private Writer.Job job;

        @Override
        public void preCheck()
        {
            this.init();
            this.job.writerPreCheck(this.originalConfig);
        }

        //@Override
        public void init()
        {
            this.originalConfig = super.getPluginJobConf();
            this.job = new Writer.Job();
            this.job.init(this.originalConfig);
        }

        // 一般来说，是需要推迟到 task 中进行pre 的执行（单表情况例外）
        @Override
        public void prepare()
        {
            //实跑先不支持 权限 检验
            //this.job.privilegeValid(this.originalConfig, DATABASE_TYPE);
            this.job.prepare(this.originalConfig);
        }

        @Override
        public List<Configuration> split(int mandatoryNumber)
        {
            return this.job.split(this.originalConfig, mandatoryNumber);
        }

        // 一般来说，是需要推迟到 task 中进行post 的执行（单表情况例外）
        @Override
        public void post()
        {
            this.job.post(this.originalConfig);
        }


        public void destroy()
        {
            this.job.destroy(this.originalConfig);
        }

    }

    public static class Task extends common.spi.Writer.Task
    {
        private Configuration writerSliceConfig;
        private Writer.Task writerTask;


        public void init()
        {
            this.writerSliceConfig = super.getPluginJobConf();
            this.writerTask = new Writer.Task();
            this.writerTask.init(this.writerSliceConfig);
        }

        @Override
        public void prepare()
        {
            this.writerTask.prepare(this.writerSliceConfig);
        }

        //TODO 改用连接池，确保每次获取的连接都是可用的（注意：连接可能需要每次都初始化其 session）
        public void startWrite(Consume consume)
        {
            this.writerTask.startWrite(consume, this.writerSliceConfig,
                    super.getTaskPluginCollector());
        }

        @Override
        public void post()
        {
            this.writerTask.post(this.writerSliceConfig);
        }


        public void destroy()
        {
            this.writerTask.destroy(this.writerSliceConfig);
        }

        @Override
        public boolean supportFailOver()
        {
            String writeMode = writerSliceConfig.getString(Key.write_mode);
            return "replace".equalsIgnoreCase(writeMode);
        }

    }


}
